package com.qingcloud.df.executor.config;

import com.google.common.collect.Lists;
import com.qingcloud.df.executor.cluster.ServerActorHandler;
import com.qingcloud.df.executor.worker.ExecutorActorHandler;
import com.qingcloud.df.sdk.component.config.ExecutorConfig;
import com.qingcloud.df.sdk.component.enums.ServerCatagory;
import com.qingcloud.df.sdk.component.pojo.AkkaAddress;
import com.qingcloud.df.sdk.component.transport.engine.AkkaEngineConfig;
import com.qingcloud.df.sdk.component.transport.engine.RemoteExeEngine;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.stereotype.Component;

import javax.annotation.Resource;

/**
 * akka 初始化
 */

@Component
public class AkkaConfig {

    @Resource
    ExecutorActorHandler executorActorHandler;
    @Resource
    ServerActorHandler serverActorHandler;

    private static RemoteExeEngine remoteExeEngine;
    public void init() {
        Logger logger = LoggerFactory.getLogger(AkkaConfig.class);
        AkkaEngineConfig engineConfig = null;
        String type =  ExecutorConfig.getType();
        String[] nodeTypes =  type.split("#");
        if(nodeTypes!=null && nodeTypes.length>0){
            for(String nodeType : nodeTypes){
                if(nodeType.equalsIgnoreCase(ServerCatagory.SERVER.toString())) {
                    engineConfig = new AkkaEngineConfig()
                            .setServerCatagory(ServerCatagory.SERVER)
                            .setBindAkkaAddress(AkkaAddress.fromIpv4(ExecutorConfig.getAkkaServerAddress()))
                            .setActorList(Lists.newArrayList(serverActorHandler));
                }else{
                    engineConfig = new AkkaEngineConfig()
                            .setServerCatagory(ServerCatagory.EXECUTOR)
                            .setBindAkkaAddress(AkkaAddress.fromIpv4(ExecutorConfig.getAkkaNodeAddress()))
                            .setActorList(Lists.newArrayList(executorActorHandler));
                }
                remoteExeEngine = new RemoteExeEngine();
                remoteExeEngine.start(engineConfig);
            }
        }
    }

    public void close(String type){
        String[] nodeTypes =  type.split("#");
        for(String nodeType : nodeTypes) {
            if (nodeType.equalsIgnoreCase(ServerCatagory.SERVER.toString())) {
                RemoteExeEngine.close(ServerCatagory.SERVER);
            }else{
                RemoteExeEngine.close(ServerCatagory.EXECUTOR);
            }
        }
    }
}
